package rocketmq

import (
	"context"
	"fmt"
	rmq_client "github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
	rmqProto "github.com/apache/rocketmq-clients/golang/v5/protocol/v2"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

const (
	ConsumerFIFOGroup = "test_fifo_consumer"
)

func FIFOConsumer() {

	// log to console
	os.Setenv("mq.consoleAppender.enabled", "true")
	rmq_client.ResetLogger()
	// In most case, you don't need to create many consumers, singleton pattern is more recommended.
	simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
		Endpoint:      Endpoint,
		ConsumerGroup: ConsumerFIFOGroup,
		Credentials: &credentials.SessionCredentials{
			AccessKey:    AccessKey,
			AccessSecret: SecretKey,
		},
	},
		rmq_client.WithAwaitDuration(awaitDuration),
		rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
			TopicFIFO: rmq_client.SUB_ALL,
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	// start simpleConsumer
	err = simpleConsumer.Start()
	if err != nil {
		log.Fatal(err)
	}
	// graceful stop simpleConsumer
	defer func() {
		if r := recover(); r != nil {
			fmt.Println(r)
		}
		_ = simpleConsumer.GracefulStop()
	}()
	fmt.Println("start test_dealy_consumer receive message")
	// Each Receive call will only select one broker queue to pop messages.
	// Enable multiple consumption goroutines to reduce message end-to-end latency.
	ch := make(chan struct{})
	wg := &sync.WaitGroup{}
	for i := 0; i < receiveConcurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ch:
					return
				default:
					mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)

					errRpcStatus, ok := rmq_client.AsErrRpcStatus(err)
					if ok {
						errCode := rmqProto.Code(errRpcStatus.GetCode())
						switch errCode {
						case rmqProto.Code_MESSAGE_NOT_FOUND: //没有更多消息
							fmt.Println("receive message error: " + err.Error())
							time.Sleep(time.Second * 3)
						default:
							if err != nil {
								fmt.Println("receive message error: " + err.Error())
							}
						}
					}
					// ack message
					for _, mv := range mvs {
						fmt.Println(mv)
						fmt.Println("mesaage body" + string(mv.GetBody()))
						if err := simpleConsumer.Ack(context.TODO(), mv); err != nil {
							fmt.Println("ack message error: " + err.Error())
						}
					}
				}
			}
		}()
	}
	exit := make(chan os.Signal, 1)
	signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
	// wait for exit
	<-exit
	close(ch)
	wg.Wait()

}
